package net.lzzy.udfa;

import org.apache.spark.sql.*;
import org.apache.spark.sql.expressions.MutableAggregationBuffer;
import org.apache.spark.sql.expressions.UserDefinedAggregateFunction;
import org.apache.spark.sql.types.*;

import java.util.Arrays;
import java.util.Properties;

public class UdfaTest2 {
    public static void main(String[] args) throws AnalysisException {
        SparkSession ss = SparkSession.builder().appName(UdfaTest2.class.getName()).master("local").getOrCreate();

        //连接数据库Mysql
        Properties properties = new Properties();
        properties.setProperty("user", "root");
        properties.setProperty("password", "123456");
        Dataset<Row> stuInfo = ss.read().jdbc("jdbc:mysql://127.0.0.1:3306/student?serverTimezone=GMT%2B8", "tab_studentinfo", properties);

        //定义一个自定义聚合
        ss.udf().register("my_max",new MyMax());
        ss.udf().register("my_contains",new MyContains());
        //读取json文件
        Dataset<Row> jsonFile = ss.read().json("src/newStudentInfo6.json");
        //设置一个视图
        jsonFile.createTempView("student");
        //使用sql语句查询出来参数
        Dataset<Row> sqlSelect = ss.sql("select name,sex,age,phone,school,my_contains(`major`) over(partition by school) as `age` from student");
//        sqlSelect.show();
        //把查询出来的数据传入数据库当中
        sqlSelect.write()
                .mode(SaveMode.Append)
                .jdbc("jdbc:mysql://127.0.0.1:3306/student?serverTimezone=GMT%2B8", "tab_student_major", properties);

    }

    public static class MyMax extends UserDefinedAggregateFunction{

        @Override
        public StructType inputSchema() {
            return DataTypes.createStructType(Arrays.asList(new StructField("age",DataTypes.StringType,true,Metadata.empty())));
        }

        @Override
        public StructType bufferSchema() {
            return DataTypes.createStructType(Arrays.asList(new StructField("age",DataTypes.StringType,true,Metadata.empty())));
        }

        @Override
        public DataType dataType() {
            return DataTypes.StringType;
        }

        @Override
        public boolean deterministic() {
            return true;
        }

        @Override
        public void initialize(MutableAggregationBuffer buffer) {
            buffer.update(0,"大数据");
        }

        @Override
        public void update(MutableAggregationBuffer buffer, Row row) {
                int aLong = Integer.parseInt(buffer.getString(0));
                int i = Integer.parseInt(row.getString(0));
                if (i>=aLong){
                    buffer.update(0,String.valueOf(i));
                }
        }

        @Override
        public void merge(MutableAggregationBuffer buffer, Row row) {
                int aLong = Integer.parseInt(buffer.getString(0));
                int i = Integer.parseInt(row.getString(0));
                if (i>=aLong){
                    buffer.update(0,String.valueOf(i));
                }
        }

        @Override
        public Object evaluate(Row row) {
            return row.getString(0);
        }
    }

    public static class MyContains extends UserDefinedAggregateFunction{

        @Override
        public StructType inputSchema() {
            return DataTypes.createStructType(Arrays.asList(new StructField("age",DataTypes.StringType,true,Metadata.empty())));
        }

        @Override
        public StructType bufferSchema() {
            return DataTypes.createStructType(Arrays.asList(new StructField("age",DataTypes.StringType,true,Metadata.empty())));
        }

        @Override
        public DataType dataType() {
            return DataTypes.StringType;
        }

        @Override
        public boolean deterministic() {
            return true;
        }

        @Override
        public void initialize(MutableAggregationBuffer buffer) {
            buffer.update(0,"大数据");
        }

        @Override
        public void update(MutableAggregationBuffer buffer, Row row) {
            String bufferStr = buffer.getString(0);
            String rowStr = row.getString(0);
            boolean flag = rowStr.contains(bufferStr);
            if (flag){
                buffer.update(0,rowStr);
            }

        }

        @Override
        public void merge(MutableAggregationBuffer buffer, Row row) {
            String bufferStr = buffer.getString(0);
            String rowStr = row.getString(0);
            boolean flag = rowStr.contains(bufferStr);
            if (flag){
                buffer.update(0,rowStr);
            }
        }

        @Override
        public Object evaluate(Row row) {
            return row.getString(0);
        }
    }
}
